package my.thesis.bolts.persistors;

import java.sql.Connection;
import java.sql.DriverManager;
import java.text.DateFormat;
import java.text.Format;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.sql.Statement;

import my.thesis.analytics.metrics.LastTimePersistMetric;
import my.thesis.analytics.metrics.ThroughputMetric;

import org.joda.time.DateTime;

import redis.clients.jedis.Jedis;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Tuple;

public class HashStreamPersistor extends BaseRichBolt 
{
	private static final long serialVersionUID = 1L;
	private Jedis jedis;
	private OutputCollector collector;
	Connection connect=null;
	Integer timeIntervalSec;
	
	transient LastTimePersistMetric _lastTimeMetric;
	private static Boolean metricsOn=false;
	
	private String database;
	
	public void prepare(Map stormConf, TopologyContext context,OutputCollector collector) 
	{
		//jedis=new Jedis(this.database);
		jedis=new Jedis(database,6379,60);
		
		jedis.select(2);
		this.collector=collector;
		
		if(metricsOn)
		{
			_lastTimeMetric=new LastTimePersistMetric();
			context.registerMetric("topHashtagPersist", _lastTimeMetric, 40);
			_lastTimeMetric.setStartTime();
		}
	}
	
	public HashStreamPersistor(Integer ti,String connectionString)
	{
		this.timeIntervalSec=ti;
		this.database=connectionString;
	}
		
	public void declareOutputFields(OutputFieldsDeclarer declarer) 
	{
		// TODO Auto-generated method stub
	}
	
	public void execute(Tuple tuple) 
	{	
		if(metricsOn)
		{
			_lastTimeMetric.setTimeReceived();
		}
		
		Long timeInterval=tuple.getLongByField("timeInterval");

		@SuppressWarnings("unchecked")
		HashMap <String, HashMap<String, Integer>> countsForInterval = (HashMap <String, HashMap<String, Integer>>) tuple.getValueByField("countsForInterval");
			
		//TEST
		//System.out.println("Persisting top hashtags...");

		for(Entry<String, HashMap<String, Integer>> entry : countsForInterval.entrySet())	
		{
			 //-------------- CONCURRENT ----------------------
			
			 String lang = entry.getKey();
			 			 			 
			 Long timeExactMillis=timeInterval * timeIntervalSec * 1000;
			 DateTime timeExact=new DateTime(timeExactMillis);
			 
			 String timeIntervalLang = timeExact.toString() + " - " + lang;

			 Set<String> keys = new HashSet<String>();
			 
			 for(String h : entry.getValue().keySet())
			 {
				 String hn=new String(h);
				 keys.add(hn);
			 }
			 
			 for(String h : keys)
			 {
					 jedis.hset(timeIntervalLang,h,entry.getValue().get(h).toString());
			 }
			 
			//-------------- CONCURRENT ----------------------
			
		  }	
	}
				
	public void cleanup()
	{
	    if (jedis.isConnected()) 
	    {
		      jedis.quit();
		}	
	}

}
